package org.example;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.Message;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.MessageListener;
import org.apache.activemq.ActiveMQConnectionFactory;

import java.util.ArrayList;
import java.util.Scanner;

class MyListenerVariance implements MessageListener {

    int N;
    int num = 0;
    ArrayList arrayList = new ArrayList();

    MyListenerVariance(int N) {this.N = N;}
    public void onMessage(Message message) {
        try {
            double sum = 0;
            double sum_var = 0;
            String s = ((TextMessage)message).getText();
            double signal = Double.valueOf(s);
            num++;
            arrayList.add(signal);
            if(num >= N) {
                for(int i = num - N; i < num; i++) {
                    double d = (double)arrayList.get(i);
                    sum += d;
                }
                for(int i = num - N; i < num; i++) {
                    double d = (double)arrayList.get(i);
                    sum_var += (d - sum/N) * (d - sum/N);
                }
                System.out.println("方差：" + sum_var/N);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

public class CounterVariance {

    public static void main(String[] args) throws JMSException {
        String brokerURL = "tcp://localhost:61616";
        ConnectionFactory factory = null;
        Connection connection = null;
        Session session = null;
        Topic topic = null;
        MessageConsumer messageConsumer = null;
        MyListenerVariance listener = null;


        try {
            factory = new ActiveMQConnectionFactory(brokerURL);
            connection = factory.createConnection();

            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            topic = session.createTopic("MYTOPIC");

            messageConsumer = session.createConsumer(topic);

            System.out.println("请输入N的值：");
            Scanner scanner = new Scanner(System.in);
            int N = scanner.nextInt();

            listener = new MyListenerVariance(N);

            messageConsumer.setMessageListener(listener);

            connection.start();

            System.out.println("Press any key to exit.");
            System.in.read();   // Pause
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            connection.close();
        }
    }
}
